package com.hyf.kafka.demo;


import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

public class KafKaProducer extends Thread{

    private final Producer<Integer,String> producer;
    private final String topic;
    private final Properties properties=new Properties();

    public KafKaProducer(String topic) {
        this.topic = topic;
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        properties.put("metadata.broker.list","47.93.157.178:19092");
        producer=new Producer<Integer, String>(new ProducerConfig(properties));

    }

    @Override
    public void run() {
        int messageNo=1;
        while (true){
            String messageStr=new String("Message_"+messageNo);
            System.out.println("Send:"+messageStr);
            producer.send(new KeyedMessage<Integer, String>(topic,messageStr));
            messageNo++;
            try {
                sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
